package com.xncoding.rocketmq.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.List;

/***
* @description Canal的TCP模式（生产环境版）更多配置场景请移步{@link SimpleCanalClientExample}
* @link {https://github.com/alibaba/canal/wiki/ClientExample}
* @author luming
* @date 2021/9/10 11:44
* @throws
*/
@Slf4j
public class ProdCanalClientExample {


    public static void main(String args[]) {
        // 创建连接
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1",
                11111), "api3", "", "");
        int batchSize = 100;
        try {
            connector.connect();
            /**
             * 不提交客户端filter，以服务端的filter为准 {@link CanalConnector#subscribe() }
             * 提交客户端filter，将覆盖服务端的filter，因此会影响到其他客户端，目前无法做到多个客户端消费不同的表 {@link CanalConnector#subscribe(String)}
             * 官方说明：
             * TODO: 后续可以考虑，如果本次提交的filter不为空，在执行过滤时，是对canal server filter + 本次filter的交集处理，达到只取1份binlog数据，多个客户端消费不同的表
             */
//            connector.subscribe(".*\\..*");
            connector.subscribe("fxq-api.t_sensitive_word");
            connector.rollback();
            log.info("敏感词变动监听中...");
            while (true) {
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    try {
                        Thread.sleep(10000);
                    } catch (InterruptedException e) {
                    }
                } else {
                    //过滤掉事务开启和关闭两种类型！！
                    boolean b = message.getEntries().stream()
                            .anyMatch(entry ->
                                    entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONBEGIN && entry.getEntryType() != CanalEntry.EntryType.TRANSACTIONEND);
                    if (b) {
                        rebuildCache();
                    }
                }

                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }
        } catch (Exception e){
            log.error("敏感词变动监听启动失败！",e);
        } finally {
            connector.disconnect();
        }
    }

    /**
     * 缓存重建
     */
    private static void rebuildCache(){
        log.info("相关业务开始执行...");
    }
}